package com.lee.mq.mqclient;

import com.lee.mq.common.*;

import java.io.*;
import java.net.MalformedURLException;
import java.net.Socket;
import java.net.SocketException;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: lishuo
 * Date: 2023-10-31
 * Time: 11:40
 */
public class Connection {
    // 每个connection有一个socket
    private Socket socket = null;
    // 一个connection管理多个channel，这里使用哈希表管理起来
    private ConcurrentHashMap<String,Channel> channelMap = new ConcurrentHashMap<>();

    private InputStream inputStream = null;
    private OutputStream outputStream = null;
    private DataInputStream dataInputStream = null;
    private DataOutputStream dataOutputStream = null;

    private ExecutorService executorService = null;

    // 在构造方法中进行连接的初始化
    public Connection(String host,int post) throws IOException {
        socket = new Socket(host,post);
        inputStream = socket.getInputStream();
        outputStream = socket.getOutputStream();
        dataInputStream = new DataInputStream(inputStream);
        dataOutputStream = new DataOutputStream(outputStream);

        executorService = Executors.newFixedThreadPool(4);

        // 创建一个扫描线程负责不断的从socket中读取响应数据，交给对应的channel处理
        Thread t = new Thread(()->{
            try {
                while (!socket.isClosed()) {
                    Response response = readResponse();
                    dispatchResponse(response);
                }
            } catch(SocketException e) {
                System.out.println("[Connection] 连接正常断开!");
            } catch (IOException | ClassNotFoundException | MqException e) {
                // 连接异常关闭
                System.out.println("[Connection] 连接异常断开!");
                e.printStackTrace();
            } finally {
                try {
                    close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
        t.start();
    }

    // 使用这个方法来分别处理, 当前的响应是一个针对控制请求的响应, 还是服务器推送的消息.
    private void dispatchResponse(Response response) throws IOException, ClassNotFoundException, MqException {
        if (response.getType() == 0xc) {
            // 0xc 表示当前响应时服务器推送的消息
            // (1)进行消息解析
            SubScribeReturns subScribeReturns = (SubScribeReturns) BinaryTool.toObject(response.getPayload());
            // (2)根据channelId或consumerTag找到对应的channel对象
            Channel channel = channelMap.get(subScribeReturns.getChannelId());
            if (channel == null) {
                throw new MqException("[Connection] 消息对应的 channel 在客户端中不存在 channel = "+subScribeReturns.getChannelId());
            }
            // (3)执行channel对象内部的回调
            executorService.submit(()->{
                try {
                    channel.getConsumer().handleDelivery(subScribeReturns.getConsumerTag(),subScribeReturns.getBasicProperties(),
                            subScribeReturns.getBody());
                } catch (IOException | MqException e) {
                    e.printStackTrace();
                }
            });
        } else {
            // 0x1~0xb表示当前响应是一个针对控制请求的响应
            // (1)进行响应解析
            BasicReturns basicReturns = (BasicReturns) BinaryTool.toObject(response.getPayload());
            // (2)根据channelId找到对应channel对象
            Channel channel = channelMap.get(basicReturns.getChannelId());
            if (channel == null) {
                throw new MqException("[Connection] 消息对应的 channel 在客户端中不存在 channel = "+basicReturns.getChannelId());
            }
            // (3)将响应结果放到对应channel的哈希表中
            channel.putReturns(basicReturns);
        }
    }

    // 读取响应
    public Response readResponse() throws IOException {
        int type = dataInputStream.readInt();
        int length = dataInputStream.readInt();
        byte[] payload = new byte[length];
        int actualLen = dataInputStream.read(payload);
        System.out.println("type = "+type+"载荷预期长度 = "+length+",实际读取长度 = "+actualLen);
        if (actualLen != length) {
            throw new IOException("[Connection] 读取数据不完整!");
        }
        Response response = new Response();
        response.setType(type);
        response.setLength(length);
        response.setPayload(payload);
        System.out.println("[Connection] 收到响应: type = "+type+",length = "+length);
        return response;
    }

    // 发送请求BUG003(写网卡需要flush)
    synchronized public void writeRequest(Request request) throws IOException {
        dataOutputStream.writeInt(request.getType());
        dataOutputStream.writeInt(request.getLength());
        dataOutputStream.write(request.getPayload());
        // ***
        dataOutputStream.flush();
        System.out.println("[Connection] 发送请求: type = "+request.getType()+",length = "+request.getLength());
    }

    // 关闭connection连接，释放资源
    public void close() throws IOException {
        executorService.shutdownNow();
        channelMap.clear();
        inputStream.close();
        outputStream.close();
        socket.close();
    }

    /**
     * 在 Connection 中创建一个 channel
     * BUG002(顺序问题：先记录在通知服务器)
     */
    public Channel createChannel() throws IOException {
        // 生成channelId
        String  channelId = "C-"+UUID.randomUUID().toString();
        Channel channel = new Channel(channelId,this);

        // 将channel记录到当前连接的哈希表中
        channelMap.put(channelId,channel);

        // 调用channel内部方法在服务器端创建channel
        boolean ok = channel.createChannel();
        if (!ok) {
            // 从connection中的哈希表中删除
            channelMap.remove(channelId);
            return null;
        }

        return channel;
    }


}
